use asse::asse_proc;
use base::Result;
use base::{
    delete_all_tmp_table, get_back_date, get_config, get_etl_redis_cli, get_imp_config,
    get_sys_date, init_app_config, init_imp_config, init_join_config, load_feqp,
    source_begin_import, source_finish_import, SourceType,
};
use big_table::{
    calc_proc, calc_proc_with_ref, get_big_table_sqls, init_big_table, is_big_table_init,
};
use chrono::{Days, NaiveDate};
use data_import::{import_one_file, import_one_source};
use divi_back::divi_back;
use ftp::FtpDownload;
use redis::Commands;
use strum::IntoEnumIterator;
mod asse; //考核结果
mod base; // 基础工具
mod big_table; //大宽表
mod calc; //计算
mod data_import; //数据库操作
mod data_proc; //执行sql
mod divi_back; //认定关系处理
mod ftp; //ftp连接
mod grpc; //grpc服务
pub use base::get_work_path;
use clap::{arg, command, ArgAction, Command};
use cron::Schedule;
use std::str::FromStr;
use tokio::time::Instant;
///初始化配置
pub async fn init() -> Result<()> {
    if is_big_table_init() {
        return Ok(());
    }
    init_app_config().await?;
    init_join_config().await?;
    init_imp_config().await?;
    init_big_table()?;
    Ok(())
}
/// etl运行
pub async fn etl_run(date: Option<NaiveDate>) -> Result<()> {
    let rcil = get_etl_redis_cli();
    let mut rcon = rcil.get_connection()?;
    let res: bool = rcon.set_nx("import_doing", 1_u8)?;
    if res == false {
        log::warn!("etl程序正在运行中，本次运行终止");
        return Ok(());
    }
    let imp_err: bool = rcon.get("import_error")?;
    if imp_err {
        log::error!("上次etl程序运行失败，本次运行终止，请检查并处理问题");
        return Ok(());
    }
    let _: () = rcon.expire("import_doing", 60 * 60 * 8)?;
    let mut result: base::Result<()> = Ok(());
    let mut exec_date: Option<NaiveDate> = None;
    for st in SourceType::iter() {
        let date = if let Some(date) = date {
            date
        } else {
            let sys_date = get_sys_date(st).await?;
            sys_date.checked_add_days(Days::new(1)).unwrap()
        };
        result = import_one_source(date, st, false).await;
        if result.is_err() {
            break;
        }
        if let Some(ed) = exec_date {
            if ed > date {
                exec_date = Some(date);
            }
        } else {
            exec_date = Some(date);
        }
    }
    if let Some(date) = exec_date {
        calc_proc(&date, false).await?;
        let _: () = rcon.del("import_doing")?;
        delete_all_tmp_table().await?;
    }
    if let Err(e) = result {
        log::error!("etl程序运行失败，错误信息：{}", e);
        let _: () = rcon.set_ex("import_error", 1, 60 * 60 * 8)?;
        let _: () = rcon.del("import_doing")?;
        return Err(e);
    }
    Ok(())
}
/// 数据源导入
pub async fn source_import(date: NaiveDate, source_type: SourceType, is_again: bool) -> Result<()> {
    import_one_source(date, source_type, is_again).await?;
    delete_all_tmp_table().await?;
    Ok(())
}
/// 文件导入
pub async fn file_import(
    date: NaiveDate,
    source_type: SourceType,
    file_name: &str,
    is_again: bool,
) -> Result<()> {
    let ic = get_imp_config(source_type);
    let file_name_prefix = ic.file_name_prefix.as_str();
    let file_ext_name = ic.file_ext_name.as_str();
    let confirm_file_ext_name = ic.confirm_file_ext_name.as_str();
    let coded_format = &ic.coded_format;
    let tconf = ic
        .tables
        .iter()
        .find(|t| t.file_name == file_name)
        .ok_or(anyhow::anyhow!(
            "在数据源配置{}中未找到文件{}的配置",
            source_type,
            file_name
        ))?;
    let mut ftp = FtpDownload::from_conf(&ic.ftp)?;
    if tconf
        .db_calc_cols
        .iter()
        .any(|(_, v)| v.contains("JOIN_FEQP_CURR"))
    {
        let count = load_feqp(&date).await?;
        if file_name != "f_cor_abpfk0" && count == 0 {
            let date_str = date.format("%Y%m%d").to_string();
            return Err(anyhow::anyhow!("{date_str}日的汇率表为空，请联系管理员").into());
        }
    }
    source_begin_import(&date, &source_type).await?;
    import_one_file(
        date,
        source_type,
        file_name_prefix,
        file_ext_name,
        confirm_file_ext_name,
        ic.column_separator,
        coded_format,
        tconf,
        &mut ftp,
        is_again,
    )
    .await?;
    for dfc in tconf.divi_fact_convs.iter() {
        dfc.divi_type.drop_table(&date).await?;
    }
    source_finish_import(&date, &source_type, true).await?;
    let sis = get_big_table_sqls(source_type, file_name);
    for si in sis {
        calc_proc_with_ref(&date, &si.col_names, false).await?;
    }
    Ok(())
}
/// 数据处理
pub async fn divi_back_proc() -> Result<()> {
    let _ = divi_back().await?;
    Ok(())
}
/// 考核结果处理
pub async fn assess_proc(back_date: Option<NaiveDate>) -> Result<()> {
    let back_date = if let Some(back_date) = back_date {
        back_date
    } else {
        let bdate = get_back_date().await?;
        bdate
    };
    asse_proc(&back_date).await?;
    delete_all_tmp_table().await?;
    Ok(())
}

pub async fn run_cli() {
    let matches = command!() // requires `cargo` feature
        .propagate_version(true)
        .subcommand_required(true)
        .arg_required_else_help(true)
        .subcommand(
            Command::new("import")
                .about("导入数据")
                .arg(arg!(-d --date <YYYYMMDD>).help("导入数据的日期,格式YYYYMMDD"))
                .arg(
                    arg!(-s --source_type <source_type>)
                        .help("数据来源类型:ODS,EDW,FTP,CAP,DLA,EDA"),
                )
                .arg(arg!(-n --name <name>).help("要导入的数据文件名，如果省略，则导入所有文件"))
                .arg(
                    arg!(-a --again <again>)
                        .help("是否重新导入，如果省略，则不重新导入")
                        .action(ArgAction::SetTrue),
                ),
        )
        .subcommand(Command::new("divi_back").about("认定关系追溯"))
        .subcommand(
            Command::new("asse_proc")
                .about("处理考核结果")
                .arg(arg!([date])),
        )
        .subcommand(Command::new("cron").about("开启进度任务,定时运行数据导入"))
        .get_matches();
    match matches.subcommand() {
        Some(("import", sub_matches)) => {
            let date = if let Some(date_str) = sub_matches.get_one::<String>("date") {
                let res = NaiveDate::parse_from_str(date_str, "%Y%m%d");
                if let Ok(date) = res {
                    date
                } else {
                    panic!("日期参数{}无效,请使用YYYYMMDD格式", date_str)
                }
            } else {
                panic!("必须指定日期参数")
            };
            let stype = if let Some(st) = sub_matches.get_one::<String>("source_type") {
                let res = SourceType::from_str(st);
                if let Ok(stype) = res {
                    stype
                } else {
                    panic!(
                        "数据源参数{}无效,请指定ODS,EDW,FTP,CAP,DLA,EDA其中的一个",
                        st
                    )
                }
            } else {
                panic!("必须指定数据源参数")
            };
            let name = if let Some(name) = sub_matches.get_one::<String>("name") {
                Some(name.as_str())
            } else {
                None
            };
            let again = if let Some(ag) = sub_matches.get_one::<bool>("again") {
                *ag
            } else {
                false
            };
            if let Err(err) = init().await {
                log::error!("{}", err.to_string());
                return;
            }
            let res = if let Some(name) = name {
                file_import(date, stype, name, again).await
            } else {
                source_import(date, stype, again).await
            };
            if let Err(err) = res {
                log::error!("{}", err.to_string());
            }
        }
        Some(("divi_back", _)) => {
            if let Err(err) = init().await {
                log::error!("{}", err.to_string());
                return;
            }
            let res = divi_back_proc().await;
            if let Err(err) = res {
                log::error!("{}", err.to_string());
            }
        }
        Some(("asse_proc", sub_matches)) => {
            if let Err(err) = init().await {
                log::error!("{}", err.to_string());
                return;
            }
            let date = if let Some(date_str) = sub_matches.get_one::<String>("date") {
                let res = NaiveDate::parse_from_str(date_str, "%Y%m%d");
                if let Ok(date) = res {
                    Some(date)
                } else {
                    panic!("日期参数{}无效,请使用YYYYMMDD格式", date_str)
                }
            } else {
                None
            };
            let res = assess_proc(date).await;
            if let Err(err) = res {
                log::error!("{}", err.to_string());
            }
        }
        Some(("cron", _)) => {
            if let Err(err) = init().await {
                log::error!("{}", err.to_string());
                return;
            }
            #[cfg(not(target_os = "windows"))]
            {
                tokio::spawn(watching_config());
            }
            let res = run_cron().await;
            if let Err(err) = res {
                log::error!("{}", err.to_string());
            }
        }
        _ => log::error!("错误的subcommand"),
    }
}
async fn run_cron() -> anyhow::Result<()> {
    let conf = get_config();
    let expression = format!(
        "{} {} {} {} {} {} {}",
        conf.cron.sec,
        conf.cron.min,
        conf.cron.hour,
        conf.cron.day_of_month,
        conf.cron.month,
        conf.cron.day_of_week,
        conf.cron.year
    );
    let schedule = Schedule::from_str(&expression).unwrap();
    let mut cur_date = chrono::Local::now();
    for dt in schedule.after(&cur_date) {
        cur_date = chrono::Local::now();
        let dura = dt - cur_date;
        let secs = dura.num_seconds();
        if secs < 0 {
            continue;
        }
        tokio::time::sleep_until(Instant::now() + std::time::Duration::from_secs(secs as u64))
            .await;
        let res = etl_run(None).await;
        if let Err(err) = res {
            log::error!("数据导入失败，错误信息{}", err)
        }
        log::logger().flush();
    }
    Ok(())
}

#[cfg(not(target_os = "windows"))]
async fn watching_config() -> anyhow::Result<()> {
    use notify::{Config, PollWatcher, RecursiveMode, Watcher};
    use std::path::Path;

    use crate::{
        base::{get_work_path, init_app_config, init_join_config, reload_imp_config},
        big_table::init_big_table,
    };
    // configure pollwatcher backend
    let config = Config::default()
        .with_compare_contents(true) // crucial part for pseudo filesystems
        .with_poll_interval(std::time::Duration::from_secs(2));
    let (tx, rx) = std::sync::mpsc::channel();
    // create pollwatcher backend
    let mut watcher = PollWatcher::new(tx, config)?;
    let path = format!("{}/config", get_work_path());
    let lo_stats = Path::new(&path).to_path_buf();
    watcher.watch(&lo_stats, RecursiveMode::Recursive)?;
    let path2 = format!("{}/sqls/big_table.sql", get_work_path());
    let lo_stats2 = Path::new(&path2).to_path_buf();
    watcher.watch(&lo_stats2, RecursiveMode::Recursive)?;
    // print all events, never returns
    for res in rx {
        match res {
            Ok(event) => {
                for path in event.paths {
                    if path.ends_with("app.yaml") {
                        if let Err(err) = init_app_config().await {
                            log::error!("{}", err)
                        } else {
                            log::info!("重新加载配置文件{}", "app.yaml")
                        }
                    } else if path.ends_with("_import.yaml") {
                        if let Err(err) = reload_imp_config(path.as_path()).await {
                            log::error!("{}", err)
                        } else {
                            log::info!(
                                "重新加载配置文件{}",
                                path.file_name().unwrap().to_str().unwrap()
                            )
                        }
                    } else if path.ends_with("joins.yaml") {
                        if let Err(err) = init_join_config().await {
                            log::error!("{}", err)
                        } else {
                            log::info!("重新加载配置文件{}", "joins.yaml")
                        }
                    } else if path.starts_with("big_table_") && path.ends_with(".sql") {
                        if let Err(err) = init_big_table() {
                            log::error!("{}", err)
                        } else {
                            log::info!(
                                "重新加载配置文件{}",
                                path.file_name().unwrap().to_str().unwrap()
                            )
                        }
                    }
                }
            }
            Err(e) => log::info!("watch error: {:?}", e),
        }
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use crate::{asse::asse_proc, base::get_back_date, divi_back_proc, file_import, init, source_import};
    use chrono::NaiveDate;
    use tracing_subscriber::{filter::LevelFilter, prelude::*};
    #[tokio::test]
    async fn test_ods_import() {
        let stdout_log = tracing_subscriber::fmt::layer().pretty();
        tracing_subscriber::registry()
            .with(stdout_log.with_filter(LevelFilter::INFO))
            .init();
        let res = init().await;
        if let Err(err) = &res {
            println!("错误信息：{}", err);
        }
        assert_eq!(res.is_ok(), true);
        let date = NaiveDate::from_ymd_opt(2022, 12, 01).unwrap();
        let res = source_import(
            date,
            crate::base::SourceType::ODS,
            false,
        )
        .await;
        if let Err(err) = &res {
            println!("错误信息：{}", err);
        }
        assert_eq!(res.is_ok(), true);
    }
    #[tokio::test]
    async fn test_file_import() {
        let stdout_log = tracing_subscriber::fmt::layer().pretty();
        tracing_subscriber::registry()
            .with(stdout_log.with_filter(LevelFilter::INFO))
            .init();
        let res = init().await;
        if let Err(err) = &res {
            println!("错误信息：{}", err);
        }
        assert_eq!(res.is_ok(), true);
        let date = NaiveDate::from_ymd_opt(2022, 02, 28).unwrap();
        let res = file_import(
            date,
            crate::base::SourceType::EDW,
            "CM_CORP_LOAN_DUBIL_SUM_M",
            true,
        )
        .await;
        if let Err(err) = &res {
            println!("错误信息：{}", err);
        }
        assert_eq!(res.is_ok(), true);
    }
    #[tokio::test]
    async fn test_asse_proc() {
        let stdout_log = tracing_subscriber::fmt::layer().pretty();
        tracing_subscriber::registry()
            .with(stdout_log.with_filter(LevelFilter::INFO))
            .init();
        let res = init().await;
        if let Err(err) = &res {
            println!("错误信息：{}", err);
        }
        assert_eq!(res.is_ok(), true);
        let res = get_back_date().await;
        if let Err(err) = &res {
            panic!("错误信息：{}", err);
        }
        let date = res.unwrap();
        let res = asse_proc(&date).await;
        if let Err(err) = &res {
            panic!("错误信息：{}", err);
        }
        assert_eq!(res.is_ok(), true);
    }
    #[tokio::test]
    async fn test_divi_proc() {
        let stdout_log = tracing_subscriber::fmt::layer().pretty();
        tracing_subscriber::registry()
            .with(stdout_log.with_filter(LevelFilter::INFO))
            .init();
        let res = init().await;
        if let Err(err) = &res {
            println!("错误信息：{}", err);
        }
        assert_eq!(res.is_ok(), true);
        let res = divi_back_proc().await;
        if let Err(err) = &res {
            panic!("错误信息：{}", err);
        }
        assert_eq!(res.is_ok(), true);
    }
}
